package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.sql.Timestamp;

public class D4WebUi {


    public static void main(String[] args) {

        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);



        String genSql = "CREATE TABLE ods_tb ( " +
                " stime TIMESTAMP(3)," +
                " name STRING," +
                " id BIGINT," +
                " ts AS localtimestamp," +
                " rowtime AS localtimestamp," +
                " WATERMARK FOR rowtime AS rowtime - interval '0' second" +
                ") WITH ( " +
                " 'connector'='datagen', " +
                " 'rows-per-second'='1'," +
                " 'fields.name.length'='1'," +
                " 'fields.id.kind'='sequence'," +
                " 'fields.id.start'='1'," +
                " 'fields.id.end'='10000000'" +
                ")";


        String sink = "CREATE TABLE pprint (" +"    " +
                "    ntime STRING, " +
                "    name STRING, " +
                "    current_watermark TIMESTAMP(3) " +
                ") WITH (" +
                "    'connector' = 'print'" +
                ")";
        String sinkSql = "INSERT INTO pprint " +
                "SELECT DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss')," +
                " name," +
                " CURRENT_WATERMARK(rowtime) AS current_watermark " +
                " FROM ods_tb";




        tableEnv.executeSql(genSql);
        tableEnv.executeSql(sink);
        tableEnv.executeSql(sinkSql);
        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
